//package com.wushi.bigdata.catalog;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import org.apache.commons.collections.CollectionUtils;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog;
//import org.apache.flink.table.api.DataTypes;
//import org.apache.flink.table.api.TableSchema;
//import org.apache.flink.table.catalog.*;
//import org.apache.flink.table.catalog.exceptions.CatalogException;
//import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
//import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
//import org.apache.flink.table.catalog.exceptions.TableNotExistException;
//import org.apache.flink.table.factories.Factory;
//import org.apache.flink.table.types.DataType;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//import java.sql.Connection;
//import java.sql.DriverManager;
//import java.sql.PreparedStatement;
//import java.sql.ResultSet;
//import java.util.*;
//
//import static org.apache.flink.table.descriptors.JdbcValidator.*;
//
///**
// * @ClassName ImpalaCatalogLocal
// * @Deacription TODO
// * @Author wushumin
// * @Date 2021-09-14 18:02
// * @Version 1.0
// **/
//public class ImpalaCatalogLocal extends AbstractJdbcCatalog {
//    private static final Logger LOG = LoggerFactory.getLogger(ImpalaCatalogLocal.class);
//    private static String builtinDatabases = "default";
//
//    static {
//        try {
//            Class.forName("com.cloudera.impala.jdbc41.Driver");
//        } catch (ClassNotFoundException e) {
//            e.printStackTrace();
//        }
//    }
//    public ImpalaCatalogLocal(String catalogName, String defaultDatabase, String baseUrl) {
//        super(catalogName, defaultDatabase, "username", "pwd", baseUrl);
//    }
//
//    @Override
//    public List<String> listDatabases() throws CatalogException {
//        List<String> pgDatabases = new ArrayList<>();
//
//        try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
//
//            PreparedStatement ps = conn.prepareStatement(String.format("select distinct table_schema from  information_schema.schema_tables"));
//            ResultSet rs = ps.executeQuery();
//
//            while (rs.next()) {
//                String dbName = rs.getString(1);
//                if (!builtinDatabases.contains(dbName)) {
//                    pgDatabases.add(rs.getString(1));
//                }
//            }
//
//            return pgDatabases;
//        } catch (Exception e) {
//            throw new CatalogException(
//                    String.format("Failed listing database in catalog %s", getName()), e);
//        }
//    }
//
//    @Override
//    public CatalogDatabase getDatabase(String databaseName)
//            throws DatabaseNotExistException, CatalogException {
//        if (listDatabases().contains(databaseName)) {
//            return new CatalogDatabaseImpl(Collections.emptyMap(), null);
//        } else {
//            throw new DatabaseNotExistException(getName(), databaseName);
//        }
//    }
//
//    // ------ tables ------
//
//    @Override
//    public List<String> listTables(String databaseName)
//            throws DatabaseNotExistException, CatalogException {
//        if (!databaseExists(databaseName)) {
//            throw new DatabaseNotExistException(getName(), databaseName);
//        }
//        List<String> tables = new ArrayList<>();
//        String dbUrl = defaultUrl.substring(0,defaultUrl.indexOf("21050")+6) + databaseName;
//        try (Connection conn = DriverManager.getConnection(dbUrl, username, pwd)) {
//
//            PreparedStatement ps = conn.prepareStatement(String.format("select * from information_schema.schema_tables where table_schema='%s'",databaseName));
//            ResultSet rs = ps.executeQuery();
//
//            while (rs.next()) {
//                String dbName = rs.getString(1);
//                    tables.add(rs.getString(2));
//            }
//            return tables;
//        } catch (Exception e) {
//            throw new CatalogException(
//                    String.format("Failed listing database in catalog %s", getName()), e);
//        }
//
//    }
//
//    @Override
//    public CatalogBaseTable getTable(ObjectPath tablePath)
//            throws TableNotExistException, CatalogException {
//        if (!tableExists(tablePath)) {
//            throw new TableNotExistException(getName(), tablePath);
//        }
//
//        try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
//            String databaseName = tablePath.getDatabaseName();
//            String tableName = tablePath.getObjectName();
//            String sql = "select c.*,t.flink_calc_options from information_schema.schema_tables t left join information_schema.schema_columns c  \n" +
//                    "on t.table_schema=c.table_schema and t.table_name=c.table_name\n" +
//                    "where t.table_schema='%s' and t.table_name='%s'";
//            PreparedStatement ps = conn.prepareStatement(String.format(sql,databaseName,tableName));
//            ResultSet rs = ps.executeQuery();
//            List<String> names = new ArrayList<>();
//            List<String> primaryNames = new ArrayList<>();
//            List<DataType> types = new ArrayList<>();
//            String options = null;
//            while (rs.next()) {
//                String columnName = rs.getString(3).trim();
//                String type = rs.getString(4).trim();
//                Integer primaryKey = rs.getInt(6);
//                Integer nullable = rs.getInt(7);
//                DataType dataType = transKUDUDataType2LogicDataType(type);
//                if(1 == primaryKey){
//                    primaryNames.add(columnName);
//                }
//                if(1 == nullable){
//                     dataType = dataType.notNull();
//                }
//                names.add(columnName);
//                types.add(dataType);
//                if(options == null){
//                    options = rs.getString(9);
//                }
//            }
//
//            TableSchema.Builder tableBuilder = new TableSchema.Builder().fields(names.toArray(new String[0]), types.toArray(new DataType[0]));
//            if(CollectionUtils.isNotEmpty(primaryNames)) {
//                tableBuilder.primaryKey(primaryNames.get(0), primaryNames.toArray(new String[0]));
//            }
//            TableSchema tableSchema = tableBuilder.build();
//
//            Map<String, String> props = new HashMap<>();
//            //取出options
//            if(StringUtils.isNotEmpty(options)){
//                JSONObject jsonObject = JSON.parseObject(options);
//                Map<String, Object> innerMap = jsonObject.getInnerMap();
//                for (Map.Entry<String, Object> entry : innerMap.entrySet()) {
//                    props.put(entry.getKey(),entry.getValue().toString());
//                }
//            }else {
//                String dbUrl = defaultUrl.substring(0,defaultUrl.indexOf("21050")+6) + databaseName;
//                //默认为JDBCconnector
////                props.put(CONNECTOR.key(), IDENTIFIER);
//                props.put("connector.type","jdbc");
//                props.put(CONNECTOR_DRIVER,"com.cloudera.impala.jdbc41.Driver");
//                props.put(CONNECTOR_URL, dbUrl);
//                props.put(CONNECTOR_TABLE, tablePath.getObjectName());
//                props.put(CONNECTOR_USERNAME, username);
//                props.put(CONNECTOR_PASSWORD, pwd);
//            }
//            return new CatalogTableImpl(tableSchema, props, "");
//        } catch (Exception e) {
//            throw new CatalogException(
//                    String.format("Failed get table %s", tablePath.getFullName()), e);
//        }
//    }
//
//
//    /**
//     * 将kudu数据类型转为flink数据类型
//     *
//     * @param columnType
//     * @return
//     */
//    private static DataType transKUDUDataType2LogicDataType(String columnType) {
//        String columnTypeStr = columnType.replaceAll("\\s", "").replaceAll("\\(.*\\)", "").toLowerCase();
//
//        //        int precision = metadata.getPrecision(colIndex);
//        //        int scale = metadata.getScale(colIndex);
//
//
//        switch (columnTypeStr) {
//            case "string":
//            case "varchar":
//                return DataTypes.STRING();
//            case "bigint":
//                return DataTypes.BIGINT();
//            case "int":
//                return DataTypes.INT();
//            case "smallint":
//                return DataTypes.SMALLINT();
//            case "tinyint":
//                return DataTypes.TINYINT();
//            case "float":
//                return DataTypes.FLOAT();
//            case "double":
//                return DataTypes.DOUBLE();
////            case "decimal":  需要精确度
////                return DataTypes.DECIMAL(6,6);
//            case "boolean":
//                return DataTypes.BOOLEAN();
//            case "timestamp":
//                return DataTypes.TIMESTAMP();
//            default:
//                return null;
//        }
//    }
//
//
//    @Override
//    public boolean tableExists(ObjectPath tablePath) throws CatalogException {
//
//        List<String> tables = null;
//        try {
//            tables = listTables(tablePath.getDatabaseName());
//        } catch (DatabaseNotExistException e) {
//            return false;
//        }
//
//        return tables.contains(tablePath.getObjectName());
//    }
//
//    @Override
//    public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) throws TableAlreadyExistException, CatalogException {
//        if (tableExists(tablePath)) {
//            throw new TableAlreadyExistException(getName(), tablePath);
//        }
//        String tableName = tablePath.getObjectName();
//        String databaseName = tablePath.getDatabaseName();
////        String sql ="select * from information_schema.schema_columns c where c.table_schema='%s' and c.table_name='%s'";
//        ResolvedCatalogTable resolvedTable = null;
//        if(table instanceof ResolvedCatalogTable){
//            resolvedTable = (ResolvedCatalogTable) table;
//            ResolvedSchema schema = resolvedTable.getResolvedSchema();
//            List<Column> columns = schema.getColumns();
//            List<String> partitionKeys = schema.getPrimaryKey().isPresent() ? schema.getPrimaryKey().get().getColumns() : null;
//            Map<String, String> options = resolvedTable.getOptions();
//            String comment = resolvedTable.getComment();
//
//            saveTable(databaseName,tableName,options,comment);
//            saveColumns(databaseName, tableName, columns, partitionKeys);
//        }
//    }
//
//    private void saveColumns(String databaseName, String tableName, List<Column> columns, List<String> primaryKeys) {
//        try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
//            conn.setAutoCommit(false);
//
//            String sql="insert into `information_schema`.`schema_columns`(`table_schema`,`table_name`,`column_name`,`column_type`,`is_primary_key`,`is_not_null`,`flink_calc_expression`) values (?,?,?,?,?,?,?)";
//            PreparedStatement ps = conn.prepareStatement(sql);
//            for (Column column : columns) {
//                ps.setString(1,databaseName);
//                ps.setString(2,tableName);
//                ps.setString(3,column.getName());
//                ps.setString(4,column.getDataType().getLogicalType().asSummaryString().replaceAll("NOT NULL","").replaceAll("\\s*",""));
//                if(CollectionUtils.isNotEmpty(primaryKeys) && primaryKeys.contains(column.getName())){
//                    ps.setInt(5,1);
//                    ps.setInt(6,1);
//                }else {
//                    ps.setInt(5,0);
//                    ps.setInt(6,0);
//                }
//                if(column instanceof Column.ComputedColumn){
//                    Column.ComputedColumn column1 = (Column.ComputedColumn) column;
//                    ps.setString(7,column1.getExpression().toString());
//                }else{
//                    ps.setString(7,null);
//                }
//                ps.addBatch();
//            }
//            int[] ints = ps.executeBatch();
//            conn.commit();
//            conn.setAutoCommit(false);
//            LOG.info("======>save columns :table:{},{},result:{}",databaseName+"."+tableName,JSON.toJSONString(columns),JSON.toJSONString(ints));
//        } catch (Exception e) {
//            throw new CatalogException(
//                    String.format("Failed saveColumns in catalog %s", getName()), e);
//        }
//
//    }
//
//    private int saveTable(String databaseName, String tableName, Map<String, String> options, String comment) {
//        try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
//            conn.setAutoCommit(false);
//            String sql="insert into `information_schema`.`schema_tables`(`table_schema`,`table_name`,`table_comment`,`table_type`,`is_ods_sync`,`flink_calc_options`) values (?,?,?,?,?,?);";
//            PreparedStatement ps = conn.prepareStatement(sql);
//                    ps.setString(1,databaseName);
//            ps.setString(2,tableName);
//            ps.setString(3,comment);
//            ps.setLong(4,1);
//            ps.setLong(5,0);
//            ps.setString(6,JSON.toJSONString(options));
//            int result = ps.executeUpdate();
//            conn.setAutoCommit(true);
//            LOG.info("======>save table:{},result:{}",databaseName+"."+tableName,result);
//            return result;
//        } catch (Exception e) {
//            throw new CatalogException(
//                    String.format("Failed listing database in catalog %s", getName()), e);
//        }
//    }
//
//    @Override
//    public Optional<Factory> getFactory() {
////        return super.getFactory();
//        return Optional.empty();
//    }
//}
